package com.guojizhuang.join;

import com.guojizhuang.util.testJson;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.MultipleInputs;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class JoinDriver {
    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {

        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf,"join");

        // jar
        job.setJarByClass(JoinDriver.class);

        // mapper+输入
        // 利用MultipleInputs，实现一个Job中的多个输入文件夹，每个文件夹对应一个Mapper类
        MultipleInputs.addInputPath(
                job,
                new Path("/user/niit/data/ijcai2016_merchant_info"),
                TextInputFormat.class,
                ShopMapper.class

        );

        MultipleInputs.addInputPath(
                job,
                new Path("/user/niit/output/guo/topn2"),
                TextInputFormat.class,
                LocationMapper.class
                );

        // 虽然多个Mapper，但是只有一个Reducer，所以所有Mapper的输出类型要一致
        job.setMapOutputKeyClass(LongWritable.class);
        job.setMapOutputValueClass(Text.class);


        // reducer
        job.setReducerClass(JoinReducer.class);
        job.setOutputKeyClass(LongWritable.class);
        job.setOutputValueClass(Text.class);

        // 输出文件夹
        FileOutputFormat.setOutputPath(job,new Path("output/guo/join"));

        job.waitForCompletion(true);

        testJson.readFromHdfs("hdfs://niit-master:9000/user/niit/output/guo/join/part-r-00000");
    }
}
